// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <pollux/common/base/fs.h>
#include <pollux/testing/gtest_utils.h>
#include <pollux/common/file/file_systems.h>
#include <pollux/dwio/common/file_sink.h>
#include <pollux/common/file/temp_directory_path.h>

#include <gtest/gtest.h>

using namespace ::testing;
using namespace kumo::pollux::exec::test;

namespace kumo::pollux::dwio::common {

class LocalFileSinkTest : public testing::Test {
 protected:
  static void SetUpTestCase() {
    memory::MemoryManager::testingSetInstance({});
    pollux::filesystems::registerLocalFileSystem();
  }

  void runTest() {
    const auto root = TempDirectoryPath::create();
    const auto filePath =
        fs::path(root->getPath()) / "xxx/yyy/zzz/test_file.ext";

    ASSERT_FALSE(fs::exists(filePath.string()));

    auto localFileSink = FileSink::create(
        fmt::format("file:{}", filePath.string()), {.pool = pool_.get()});
    ASSERT_TRUE(localFileSink->isBuffered());
    localFileSink->close();

    EXPECT_TRUE(fs::exists(filePath.string()));
  }

  std::shared_ptr<pollux::memory::MemoryPool> pool_{
      memory::memoryManager()->addLeafPool()};
};

TEST_F(LocalFileSinkTest, missingRegistration) {
  POLLUX_ASSERT_THROW(runTest(), "FileSink is not registered for file:");
}

TEST_F(LocalFileSinkTest, create) {
  LocalFileSink::registerFactory();
  runTest();
}

TEST_F(LocalFileSinkTest, existFileCheck) {
  LocalFileSink::registerFactory();
  auto root = TempDirectoryPath::create();
  auto filePath = fs::path(root->getPath()) / "xxx/yyy/zzz/test_file.ext";

  auto localFileSink = FileSink::create(
      fmt::format("file:{}", filePath.string()), {.pool = pool_.get()});
  ASSERT_TRUE(localFileSink->isBuffered());
  localFileSink->close();
  EXPECT_TRUE(fs::exists(filePath.string()));

  POLLUX_ASSERT_THROW(
      FileSink::create(
          fmt::format("file:{}", filePath.string()), {.pool = pool_.get()}),
      "File exists");
}

} // namespace kumo::pollux::dwio::common
